package com.sunsys.mapreduce.enumcounter.retrieveInReducer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Counters;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;

public class RecordsCountValidationMapperDriver extends Configured implements Tool {

	@Override
	public int run(String[] arg0) throws Exception {
		String inputFile = "/home/ubuntu/workspace-info/CountersExample/input/";
		String outputPath = "/home/ubuntu/workspace-info/CountersExample/output/1";

		Configuration conf=new Configuration();
		Job job = new Job(conf, "RecordsCountValidationJob");
		
		job.setJarByClass(RecordsCountValidationMapperDriver.class);
		job.setMapperClass(RecordsCountValidationMapper.class);
		job.setReducerClass(RecordsCountValidationReducer.class);
		job.setNumReduceTasks(1);
		
		FileInputFormat.addInputPath(job, new Path(inputFile));
		FileOutputFormat.setOutputPath(job, new Path(outputPath));

		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Text.class);
		
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);

		boolean result = job.waitForCompletion(true);
		
		Counters counters = job.getCounters();
		long value = counters.findCounter(RecordsCount.FILE1_COUNTER).getValue();
		System.out.println("Counter Value:"+value);
		
		return 0;
	}
	
	public static void main(String []args) throws Exception {
		
		RecordsCountValidationMapperDriver driver = new RecordsCountValidationMapperDriver();
		driver.run(args);
		System.out.println("Thank you");
	}
	
}
